package com.fulu.game.bigdata.realtime.sink;


import java.lang

import com.alibaba.fastjson.JSONObject
import com.fulu.game.bigdata.realtime.entity.Sink
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.flink.types.Row
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.LongSerializer


class KafkaJsonSerializationSchema(sink: Sink) extends KafkaSerializationSchema[(Boolean,Row)] {

    override def serialize(element: (Boolean, Row), timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
        val result = new JSONObject()
        val row = element._2

        for(i <- 0 until sink.getFields.size() ){
            val value = row.getField(i)
            result.put(sink.getFields.get(i).name,value)
        }
        val key: Long = if(timestamp != null)  timestamp else System.currentTimeMillis()

        val keyBytes = new LongSerializer().serialize(null,key)

        val record = new ProducerRecord(sink.getKafkaTopic,keyBytes,result.toJSONString.getBytes("utf-8"))

        record
    }
}